package com.sui.bigdata.rtcadmin.service.impl;


import com.sui.bigdata.rtcadmin.constant.AppConstant;
import com.sui.bigdata.rtcadmin.exception.SparkConfException;
import com.sui.bigdata.rtcadmin.model.dto.SparkAppDto;
import com.sui.bigdata.rtcadmin.model.vo.CommonResponse;
import com.sui.bigdata.rtcadmin.repository.mapper.AppConfMapper;
import com.sui.bigdata.rtcadmin.repository.mapper.JobStatusMapper;
import com.sui.bigdata.rtcadmin.repository.model.JobStatusInfo;
import com.sui.bigdata.rtcadmin.repository.model.SparkApp;
import com.sui.bigdata.rtcadmin.service.SparkAppConfService;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;


/**
 * @Author: TingWuHuang
 * @Date: 2020/5/6 15:30
 * @description
 */
@Service
public class SparkAppConfServiceImpl implements SparkAppConfService {

    Logger logger = LoggerFactory.getLogger(SparkAppConfServiceImpl.class);

    @Autowired
    AppConfMapper appConfMapper;


    @Autowired
    JobStatusMapper jobStatusMapper;

    @Override
    public CommonResponse saveAppConfInfo(SparkAppDto sparkAppDto){
            if(StringUtils.isEmpty(sparkAppDto.getAppName())) {
                throw new SparkConfException("appName is null!");
            }
            if(StringUtils.isEmpty(sparkAppDto.getDeveloper())){
                throw new SparkConfException("developger is null!");
            }
            if(StringUtils.isEmpty(sparkAppDto.getAppResource()) || StringUtils.isEmpty(sparkAppDto.getMainClass())){
                throw new SparkConfException("spark appResource or mainClass is null!");
            }
            // 先根据appName判断是否有正在运行的任务
            SparkApp curSparkApp = appConfMapper.selectByAppName(sparkAppDto.getAppName());
            if(curSparkApp != null){
                if (AppConstant.APP_STATUS_RUNNING.equals(curSparkApp.getAppStatus()) ||
                        AppConstant.APP_STATUS_ACCEPT.equals(curSparkApp.getAppStatus())){
                    throw new SparkConfException(sparkAppDto.getAppName() + " is running,update is not support!");
                }
            }
            try {
                sparkAppDto.setAppStatus(AppConstant.APP_STATUS_EDITED);
                appConfMapper.saveAll(SparkApp.buildSparkApp(sparkAppDto));
            }catch (Exception e){
                logger.error("appConfInfo save exception: {}", e);
                throw new SparkConfException("appConfInfo save error!");
            }

        return CommonResponse.Builder.buildSuccess(null);
    }

    @Override
    public CommonResponse updateAppConfInfo(SparkAppDto sparkApp) {
        String appName = sparkApp.getAppName();
        if(StringUtils.isEmpty(appName)){
            throw new SparkConfException("appName is null");
        }
        SparkApp curSparkApp = appConfMapper.selectByAppName(appName);
        if(curSparkApp == null){
            throw new SparkConfException(appName + " is not exit!");
        }

        appConfMapper.updateBasisConfByName(sparkApp.getAppName(),
                StringUtils.isNotEmpty(sparkApp.getDeveloper()) ? sparkApp.getDeveloper() : curSparkApp.getDeveloper(),
                StringUtils.isNotEmpty(sparkApp.getAlarmGroupPhone()) ? sparkApp.getAlarmGroupPhone() : curSparkApp.getAlarmGroupPhone());
        return CommonResponse.Builder.buildSuccess(null);
    }

    @Override
    public CommonResponse selectSparkConf(SparkAppDto sparkApp){
        List<SparkApp> sparkConfs = appConfMapper.selectAll(SparkApp.buildSparkApp(sparkApp));
        return CommonResponse.Builder.buildSuccess(sparkConfs);
    }


    @Override
    public CommonResponse selectStatusRecord(String appName, String appId, String engine) {
        if(StringUtils.isEmpty(appName)){
            throw new SparkConfException("appName value is invalid !");
        }

        List<JobStatusInfo> jobStatusInfos = jobStatusMapper.queryAll(appName,null, null);
        return CommonResponse.Builder.buildSuccess(jobStatusInfos);
    }


}
